17 DB연결 | ✅ 저자: 이유정(박사)

회원(User)을 등록하고, 사용자별로 물건(Item)을 등록/조회할 수 있는 작은 게시판 API를 위한 DB연동 예시입니다.

SQLAlchemy는 Python에서 데이터베이스를 다루기 위한 ORM(Object Relational Mapper)입니다. SQLAlchemy는 Python 개발자에게 "파이썬 코드로 SQL을 쓰는 능력"을 줍니다. 마치 Django에서 models.Model을 쓰듯이, SQLAlchemy도 Python 클래스를 DB 테이블로 만들어줍니다. Django는 ORM이 내장되어 있는 프레임워크이고,
FastAPI는 ORM이 없기 때문에 SQLAlchemy 같은 외부 라이브러리를 설치해서 사용하는 것이에요.

pip install sqlalchemy

FastAPI_DB/
└── sql_app/
    ├── __init__.py      # 패키지 초기화 파일
    ├── crud.py          # CRUD (Create, Read, Update, Delete) 로직
    ├── database.py      # 데이터베이스 연결 및 세션 관리
    ├── main.py          # FastAPI 애플리케이션 진입점
    ├── models.py        # SQLAlchemy ORM 모델 정의
    └── schemas.py       # Pydantic 스키마 (요청/응답 유효성 검사용)

각 파일의 역할요약:

파일명	역할 설명
__init__.py	sql_app 디렉토리를 Python 패키지로 인식시키는 역할
crud.py	데이터베이스와의 실제 CRUD 동작을 정의
database.py	DB 연결 설정, SessionLocal, engine 정의 등
main.py	FastAPI 앱 인스턴스 생성 및 라우터 등록 등
models.py	SQLAlchemy ORM 클래스로 테이블 구조 정의
schemas.py	Pydantic 모델로 요청/응답 데이터 구조 정의

📁 전체 구조 비교

FastAPI + SQLAlchemy	Django	역할 설명
main.py	views.py, urls.py	엔드포인트 정의 및 라우터 설정
models.py	models.py	데이터베이스 모델 정의 (ORM)
schemas.py	forms.py, serializers.py	입력/출력 데이터 구조 정의 및 유효성 검사
crud.py	views.py or managers.py	데이터 처리 로직 (DB 접근, 비즈니스 로직)
database.py	settings.py, apps.py	DB 연결, 세션 관리 등 설정
__init__.py	앱 폴더의 __init__.py	Python 패키지 인식용
-	admin.py, migrations/	FastAPI에선 별도로 없음 (직접 구성해야 함)

🔁 상호 작용 구조

1. 클라이언트 요청 → main.py (라우터)

• Django의 urls.py + views.py와 비슷하게,
• main.py는 FastAPI 앱 인스턴스를 생성하고,
• 특정 경로로 들어온 요청을 적절한 CRUD 함수에 연결합니다.

# main.py 
@app.post("/users/") 
def create_user(user: schemas.UserCreate):     
	return crud.create_user(db, user)

---

2. 데이터 스키마 검증 → schemas.py (Pydantic)

• Django에서 forms.py나 serializers.py처럼,
• schemas.py는 API의 입력값과 출력값을 검증하고 명세화합니다.
• 예: UserCreate, UserRead, UserInDB 등의 스키마 작성

# schemas.py 
class UserCreate(BaseModel):     
	email: str     
	password: str

---

3. 데이터 처리 로직 → crud.py

• Django에서 모델 매니저(MyModel.objects.filter(...))나 서비스 레이어 같은 역할
• 실제 DB 작업 (읽기/쓰기/수정/삭제 등)을 담당

# crud.py 
def create_user(db: Session, user: schemas.UserCreate):     
	db_user = models.User(email=user.email, hashed_pw=hash(user.password))     
	db.add(db_user)     
	db.commit()     
	db.refresh(db_user)     
	return db_user

---

4. DB 모델 정의 → models.py

• Django의 models.Model과 1:1 대응
• SQLAlchemy를 사용하여 테이블 스키마 정의

# models.py 
class User(Base):     
	__tablename__ = "users"     
	id = Column(Integer, primary_key=True)     
	email = Column(String, unique=True)

---

5. DB 연결 설정 → database.py

• Django의 settings.py에 있는 DATABASES 설정과 비슷
• SQLAlchemy의 engine, SessionLocal, Base 등을 정의

# database.py 
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" 
engine = create_engine(SQLALCHEMY_DATABASE_URL) 
SessionLocal = sessionmaker(bind=engine)

---

✅ 전체 흐름 요약 (비유 포함)

[클라이언트 요청]
   ↓
[main.py] (Django의 views.py와 urls.py 에 해당)
   ↓      
[schemas.py] ← 입력 검증 (DRF의 serializers.py/forms.py 역할)
   ↓
[crud.py] ← DB 처리 로직 (Django views.py의 API역할로 내부 DB 처리)
   ↓
[models.py] ← ORM 테이블 정의 (Django의 models.py)
   ↓
[database.py] ← DB 연결 및 세션 관리 (Django의 settings.py + DB 엔진)

---

main.py == (views.py, urls.py)

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

# 내부 모듈 불러오기 (같은 디렉토리 내의 파일들)
from . import crud, models, schemas
from .database import SessionLocal, engine

# SQLAlchemy의 모델(테이블)들을 DB에 생성
models.Base.metadata.create_all(bind=engine)

app = FastAPI() # FastAPI 애플리케이션 인스턴스 생성

# 의존성 주입 함수 - 요청마다 DB 세션을 생성하고, 종료 시 닫는다
def get_db():
    db = SessionLocal()  # DB 세션 객체 생성
    try:
        yield db  # 생성된 db를 FastAPI 내부에 전달함
        # return뒤에는 실행이 안되지만 
        # yield는 중간값을 넘겨주고 밑으로 실행합니다.
    finally:
        db.close()  # 요청 끝나면 자동으로 세션 닫음
# 위함수는 의존성 주입에 사용할 수 있는 함수로 정의한 것일 뿐이고,  
# 실제로 Depends(get_db)로 쓰일 때 비로소 의존성 주입이 됩니다.

models.Base.metadata.create_all(bind=engine) 
# 이 한 줄은 FastAPI와 SQLAlchemy 프로젝트에서 데이터베이스에 테이블을 
# 실제로 생성해주는 중요한 코드입니다.
# Base가 가지고 있는 모든 테이블 정의(metadata)를 engine에 연결된 데이터베이스에 적용해줘라는 뜻

# 사용자 생성 API
@app.post("/users/", response_model=schemas.User)
# 요청은 schemas.UserCreate를 따르고, 응답은 schemas.User 형식으로 반환한다는 뜻

def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    """
    사용자 생성 엔드포인트
    - 이미 존재하는 이메일이면 400 에러 반환
    - 존재하지 않으면 DB에 사용자 추가
    """
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

• FastAPI
    → schemas.UserCreate가 DRF의 입력 Serializer 같은 역할
    → schemas.User와 response_model=이 DRF의 출력 Serializer 역할
    → 자동으로 변환·검증·문서화까지 해줌

• DRF(Django REST Framework)
    → Serializer로 입력/출력 모두 처리
    → write_only=True, read_only=True 설정으로 필터링
    → 수동으로 .is_valid() 호출, .save(), .data 접근 필요

---

# 사용자 목록 조회 API
@app.get("/users/", response_model=list[schemas.User])
# 엔드포인트(`/users/`)는 사용자 테이블(models.User)에 저장된 모든 사용자 목록을 불러옵니다. 예: 이메일, 아이디, 활성화 여부 등

def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): # 의존성 함수 적용
    """
    사용자 목록을 페이지네이션하여 반환
    - skip: 앞에서 몇 명 건너뛸지
    - limit: 몇 명까지 반환할지
    """
    users = crud.get_users(db, skip=skip, limit=limit)
    return users

# 특정 사용자 조회 API
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    """
    사용자 ID로 특정 사용자 정보를 반환
    - 없으면 404 에러
    """
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

# 특정 사용자에게 아이템 추가 API
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
# schemas.ItemCreate : 요청(GET/POST)시 클라이언트가 보낼 JSON
# response_model=schemas.Item : 응답(JSON/HTML)시 서버가 보내줄 JSON

# user_id번 사용자에게 새로운 아이템(item)을 등록하는 함수
def create_item_for_user(
    user_id: int, # URL 경로에서 받은 사용자 ID
    
    item: schemas.ItemCreate, 
    # 요청 본문에서 받은 아이템 정보(title, description 등)
    
    db: Session = Depends(get_db)  # DB 연결 세션 (의존성 주입)
):
    """
    특정 사용자의 소유 아이템 추가
    - user_id를 기준으로 연결됨
    """
    return crud.create_user_item(db=db, item=item, user_id=user_id)
# 지정한 사용자에게 아이템을 만들어 DB에 저장하고, 그 결과를 돌려준다는 뜻

# 전체 아이템 목록 조회 API
@app.get("/items/", response_model=list[schemas.Item])
# GET 요청으로 /items/에 접근하면, Item 목록을 JSON으로 반환

def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    # skip: 몇 개의 데이터를 건너뛸지 (페이징 시작점)
    # limit: 최대 몇 개까지 가져올지 (한 번에 반환할 개수)
    # db: FastAPI가 get_db()를 호출해 의존성 주입으로 넘겨준 DB 세션

    """
    전체 아이템 목록을 페이지네이션하여 반환
    """
    items = crud.get_items(db, skip=skip, limit=limit)
    # 실제 DB에서 아이템들을 조회 (페이지네이션 적용)
    # GET /items/?skip=10&limit=5 이렇게 요청하면
    # 11번째 아이템부터 5개를 조회해서 반환합니다
    # 즉, SELECT * FROM items LIMIT 5 OFFSET 10과 동일한 쿼리가 실행
    # 되는 거예요.
    
    return items # 조회된 아이템 목록을 클라이언트에 JSON 형태로 응답

---

schemas.py == (forms.py, serializers.py)

from pydantic import BaseModel 
# Pydantic의 기본 모델 클래스 (데이터 검증용)를 불러옴


class ItemBase(BaseModel): 
# 아이템의 기본 정보 구조 (요청/응답 모두에서 공통 사용)

    title: str # 아이템 제목 (필수)
    description: str | None = None # 설명 (옵션, 생략 가능)


class ItemCreate(ItemBase):
# 아이템 생성 시 사용할 요청 스키마 (title, description만 필요)
    pass 
    # ItemBase와 동일하지만 구분을 위해 별도 클래스 생성 
    # (확장 가능성 있음)


class Item(ItemBase):
# ItemBase 클래스를 상속받았기 때문에, ItemBase 안에 정의된 모든 필드(`title`, `description`)가 자동으로 포함된다는 뜻입니다.

    id: int # 아이템 고유 ID (응답 전용)
    owner_id: int # 아이템을 소유한 사용자 ID (응답 전용)

    class Config:
        orm_mode = True
        # SQLAlchemy 모델을 자동으로 변환할 수 있게 설정 
        # DB 객체 → Pydantic 모델
        # DB 객체는 models.py 안에 정의된 SQLAlchemy 모델 클래스의 
        # 인스턴스를 말해요.


class UserBase(BaseModel):
# 사용자 정보에서 공통적으로 사용할 필드 정의 (email만 포함)
    email: str # 이메일 (필수 입력값)


class UserCreate(UserBase): 
# 회원가입 요청에서 사용할 입력 스키마 (email + password)
    password: str # 비밀번호 (요청 시 필요하지만 응답에는 포함되지 않음)


class User(UserBase): 
# 사용자 응답 구조 정의 (id, is_active, items 포함)

    id: int # 사용자 고유 ID (자동 생성되는 기본 키)
    is_active: bool 
    # 관리자가 회원가입한 유저를 제어하는 곳(활성화,비활성화)
    
    items: list[Item] = []
   # 사용자가 등록한 아이템 목록(Item 객체들의 리스트, 기본값은 빈 리스트)

    class Config:
        orm_mode = True
     # SQLAlchemy 모델 객체를 Pydantic 모델로 자동 변환할 수 있도록 설정

---

crud.py == (views.py)

from sqlalchemy.orm import Session
from . import models, schemas


def get_user(db: Session, user_id: int): # ID로 사용자 한 명 조회
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
# 이메일로 사용자 한 명 조회 (회원가입 시 중복 체크용)
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
# 전체 사용자 목록을 일부만 가져오기 (페이지네이션)
# skip: 앞에서 건너뛸 개수, limit: 가져올 개수
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
# 새로운 사용자를 DB에 저장
    fake_hashed_password = user.password + "notreallyhashed"
    # 실제 해싱은 아니고, 예제용으로 단순한 문자열 덧붙임
    
    db_user = models.User(email=user.email,
    # SQLAlchemy User 모델 인스턴스 생성
 hashed_password=fake_hashed_password)
 
    db.add(db_user)  # DB에 추가
    db.commit() # 실제 DB에 저장 반영
    db.refresh(db_user) 
    # 저장 후 db_user에 최신 정보 반영 (id 등 자동 생성 필드 포함)
    
    return db_user # 생성된 사용자 정보 반환


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int): # 전체 아이템 목록을 페이지네이션으로 가져옴
    db_item = models.Item(**item.dict(), owner_id=user_id)
 # Pydantic 객체(item)를 딕셔너리로 변환 후 모델에 할당, 사용자 ID도 포함
    
    db.add(db_item) # DB에 추가
    db.commit() # 저장
    db.refresh(db_item) # 반영된 내용을 다시 읽어옴 (id 등)
    return db_item  # 생성된 아이템 반환

---

models.py == (models.py)

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String  # 각 필드 타입과 외래키, 불리언 등을 정의하는 SQLAlchemy 기능들
from sqlalchemy.orm import relationship  
# 테이블 간 관계 설정을 위한 기능

from .database import Base  
# 모든 모델이 상속할 공통 Base 클래스 (declarative_base())


class User(Base):
# User 테이블 정의 (Base를 상속받아 SQLAlchemy 모델로 만듦)
    __tablename__ = "users" # 실제 DB에서 사용될 테이블 이름은 users

    id = Column(Integer, primary_key=True)                    
    # 사용자 고유 ID (자동 증가, 기본 키)
    
    email = Column(String, unique=True, index=True)           
    # 이메일 (중복 금지, 인덱스 생성)
    
    hashed_password = Column(String)                         
    # 비밀번호는 암호화된 문자열로 저장
    
    is_active = Column(Boolean, default=True)                 
    # 계정 활성화 여부 (기본값: True)

    items = relationship("Item", back_populates="owner")      
    # 이 사용자가 등록한 모든 아이템 리스트 (1:N 관계)
    # "Item.owner"와 연결된 관계 설정(User ↔ Item)
    # 1명의 사용자 → 여러 아이템


class Item(Base):                        
# Item 테이블 정의
    __tablename__ = "items"             
    # 실제 DB에서 사용될 테이블 이름은 "items"

    id = Column(Integer, primary_key=True)                    
    # 아이템 고유 ID (기본 키)
    
    title = Column(String, index=True)                        
    # 아이템 제목 (검색을 위해 인덱스 생성)
    
    description = Column(String, index=True)                  
    # 아이템 설명 (검색을 위해 인덱스 생성)

    owner_id = Column(Integer, ForeignKey("users.id"))        
    # 이 아이템을 소유한 사용자 ID (users 테이블의 id를 참조)
    
    owner = relationship("User", back_populates="items")      
    # 이 아이템의 주인 정보를 User 객체로 가져올 수 있음
    # "User.items"와 연결된 관계 설정

---

database.py == (settings.py, apps.py)

from sqlalchemy import create_engine  
# 데이터베이스에 연결하기 위한 엔진 생성 함수
from sqlalchemy.ext.declarative import declarative_base  
# 모델 클래스의 베이스(기반 클래스) 생성 함수
from sqlalchemy.orm import sessionmaker  
# DB 세션(Session)을 생성해주는 팩토리 함수


SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# 사용할 데이터베이스를 설정하는 부분 (Django의 settings.py의 DATABASES 설정과 같은 역할)
# 여기서는 SQLite를 사용하며, 현재 디렉토리에 "sql_app.db"라는 파일로 DB가 저장됨
# Django 기준으로 보면: 
#'ENGINE': 'django.db.backends.sqlite3', 
#'NAME': BASE_DIR / "db.sqlite3" 와 비슷함

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
# SQLAlchemy의 DB 엔진 생성 (실제로 DB와 연결하는 객체)
# Django에서는 settings.py의 DATABASES 설정에 따라 내부적으로 엔진이 자동 생성됨
# 여기서는 SQLite를 사용하므로, 다중 연결(thread)를 허용하려고 connect_args 설정이 필요함
# Django에서는 이런 설정을 직접 하지 않아도 되지만, FastAPI + SQLAlchemy에서는 명시적으로 설정함


SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 실제 DB 작업(조회, 저장 등)을 할 때 사용할 세션(Session)을 만들어주는 함수 (공장 함수)
# Django의 ORM에서는 내부적으로 자동으로 세션을 생성해 관리하지만,
# FastAPI + SQLAlchemy에서는 이렇게 수동으로 세션을 만들어서 직접 사용해야 함
 # - autocommit=False → 직접 commit()을 호출해야 DB에 반영됨 
 #   (Django의 save()와 비슷)
 # - autoflush=False → flush()도 수동으로 (일반적으로 False로 설정)
 # - bind=engine → 위에서 만든 DB 엔진과 연결함

Base = declarative_base()
# 모든 ORM 모델(User, Item 등)의 부모 클래스 역할
# Django에서는 models.Model을 상속받듯이,
# FastAPI에서는 Base를 상속받아 SQLAlchemy 모델 클래스를 생성함
# 이후 Base.metadata.create_all(bind=engine)으로 DB에 실제 테이블을 생성할 수 있음

경로에 맞게 실행

uvicorn sql_app.main:app --reload